package com.ermu.reactive.jdkaction;

import java.util.concurrent.Flow;

/**
 * 订阅者
 * @param <T>
 */
public class ReactiveSubscriber<T> implements Flow.Subscriber<T> {
    private String name;
    private Flow.Subscription subscription;
    final long bufferSize;
    long count;

    public String getName() {
        return name;
    }

    public Flow.Subscription getSubscription() {
        return subscription;
    }

    public ReactiveSubscriber(long bufferSize, String name) {
        this.bufferSize = bufferSize;
        this.name = name;
    }

    @Override
    public void onSubscribe(Flow.Subscription subscription) {
        //count = bufferSize - bufferSize / 2;// 当消费一半的时候重新请求
        (this.subscription = subscription).request(bufferSize);
        System.out.println("开始onSubscribe订阅");
        try {
            Thread.sleep(1000);
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    @Override
    public void onNext(T item) {
        //if (--count <= 0) subscription.request(count = bufferSize - bufferSize / 2);
        System.out.println(" ############# " + Thread.currentThread().getName()+"　name: "+name+"　item: "+item+ " ############# ");
        System.out.println(name + " received: " + item);
        try {
            Thread.sleep(10);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }

    @Override
    public void onError(Throwable throwable) {
        throwable.printStackTrace();
    }

    @Override
    public void onComplete() {
        System.out.println("Completed");
    }
}
